#ifndef PROCESSPOOL_HPP
#define PROCESSPOOL_HPP

#include <iostream>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <fcntl.h>
#include <cstring>
#include <cassert>
#include <cstdlib>
#include <cerrno>
#include <signal.h>
#include <sys/epoll.h>
#include <sys/stat.h>
#include <sys/wait.h>
using namespace std;

class process {
public:
    process() : m_pid(-1) {}

public:
    pid_t m_pid;
    int m_pipeFd[2];
};

template <typename T>
class processPool {
private:
    processPool(int listenFd, int processNumber = 8);

public:
    static processPool<T> create(int listenFd, int processNumber = 8) {
        if(!m_instance) {
            m_instance = new processPool<T>(listenFd, processNumber);
        }
        return m_instance;
    }
    ~processPool() {
        delete[] m_sub_process;
    }
    void run();

private:
    void setUpSigPipe();
    void runParent();
    void runChild();

private:
    static const int MAX_PROCESS_NUMBER = 16;
    static const int USER_PER_PROCESS = 65536;
    static const int MAX_EVENT_NUMBER = 10000;
    int m_process_number;
    int m_idx;
    int m_epollFd;
    int m_listenFd;
    int m_stop;
    process* m_sub_process;
    static processPool<T>* m_instance; 
};

template <typename T>
processPool<T>* processPool<T>::m_instance = nullptr;

static int sigPipeFd[2];

static int setNonBlocking(int fd) {
    int oldOption = fcntl(fd, F_GETFL);
    int newOption = oldOption | O_NONBLOCK;
    fcntl(fd, F_SETFL, newOption);
    return oldOption;
}

static void addFd(int epollFd, int fd) {
    epoll_event event;
    event.data.fd = fd;
    event.events = EPOLLIN | EPOLLET;
    epoll_ctl (epollFd, EPOLL_CTL_ADD, fd, &event);
    setNonBlocking(fd);
}

static void removeFd(int epollFd, int fd) {
    epoll_ctl(epollFd, EPOLL_CTL_DEL, fd, 0);
    close(fd);
}

static void sigHandler(int sig) {
    int saveErrno = errno;
    int msg = sig;
    send(sigPipeFd[1], (char*)&mag, 1, 0);
    errno = saveErrno;
}

static void addsig(int sig, void(handler)(int), bool restart = true) {
    struct sigaction sa;
    memset(&sa, 0, sizeof(sa));
    sa.sa_handler = handler;
    if(restart) {
        sa.sa_flags |= SA_RESTART;
    }
    sigfillset(&sa.sa_mask);
    assert(sigaction(sig, &sa, nullptr) != -1);
}

template <typename T>
processPool<T>::processPool(int listenFd, int processNumber) 
:   m_listenFd(listenFd), m_process_number(processNumber), m_idx(-1), m_stop(false)
{
    assert(processNumber > 0 && processNumber <= MAX_PROCESS_NUMBER);

    m_sub_process = new process[processNumber];
    assert(m_sub_process);

    for(int i = 0; i < processNumber; ++i) {
        int ret = socketpair(AF_INET, SOCK_STREAM, 0, m_sub_process[i].m_pipeFd);
        assert(ret == 0);

        m_sub_process[i].m_pid = fork();
        assert(m_sub_process[i].m_pid >= 0);
        if(m_sub_process[i].m_pid > 0) {
            close(m_sub_process[i].m_pipeFd[1]);
            continue;
        }
        else {
            close(m_sub_process[i].m_pipeFd[0]);
            m_idx = i;
            break;
        }
    }
}

template <typename T>
void processPool<T>::setUpSigPipe() {
    m_epollFd = epoll_create(5);
    assert(m_epollFd != -1);

    int ret = socketpair(AF_INET, SOCK_STREAM, 0, sigPipeFd);
    assert(ret != -1);

    setNonBlocking(sigPipeFd[1]);
    addFd(epollFd, sigPipeFd[0]);

    addsig(SIGCHLD, sigHandler);
    addsig(SIGTERM, sigHandler);
    addsig(SIGINT, sigHandler);
    addsig(SIGPIPE, SIG_IGN);
}

template <typename T>
void processPool<T>::run() {
    if(m_idx != -1) {
        runChild();
        return;
    }
    runParent();
}

template <typename T>
void processPool<T>::runChild() {
    setUpSigPipe();

    int pipeFd = m_sub_process[m_idx].pipeFd[1];
    addFd(m_epollFd, pipeFd);

    epoll_event events[MAX_EVENT_NUMBER];
    T* users = new T[USER_PER_PROCESS];
    assert(users);
    int number = 0;
    int ret = -1;

    while(!m_stop) {
        number = epoll_wait(m_epollFd, events, MAX_EVENT_NUMBER, -1);
        if(number < 0 && errno != EINTR) {
            cout << "epoll failure" << endl;
            break;
        }

        for(int i = 0; i < number; ++i) {
            int sockFd = events[i].data.fd;
            if(sockFd == pipeFd && events[i].events & EPOLLIN) {
                int client = 0;
                ret = recv(sockFd, (char*)&client, sizof(client), 0);
                if((ret < 0 && errno != EAGAIN) || ret == 0) {
                    continue;
                }
                else {
                    struct sockaddr_in clientAddress;
                    socklen_t clientAddrLen = sizeof(clientAddress);
                    int connFd = accept(m_listenFd, (struct address*)&clientAddress, &clientAddrLen);
                    if(connFd < 0) {
                        cout << "errno is: " << errno << endl;
                        continue;
                    }
                    addFd(m_epollFd, connFd);
                    users[connFd].init(m_epollFd, connFd, clientAddress);
                }
            }
            else if (sockFd == sigPipeFd[0] && events[i].events * EPOLLIN) {
                int sig;
                char signals[1024];
                ret = recv(sigPipeFd[0], signals, sizof(signals), 0);
                if(ret <= 0) {
                    continue;
                }
                else {
                    for(int i = 0; i < ret; ++i) {
                        switch(signals[i]) {
                            case SIGCHLD: {
                                // 接收到了子进程退出的信号，这里应该由父进程执行回收
                                pid_t pid;
                                int stat;
                                while(pid = waitpid(-1, &stat, WNOHANG) > 0) {
                                    continue;
                                }
                                break;
                            }
                            case SIGTERM:
                            case SIGINT: {
                                m_stop = true;
                                break;
                            }
                            default: {
                                break;
                            }
                        }
                    }
                }
            }
            else if (events[i].events & EPOLLIN){
                users[sockFd].process();
            }
            else {
                continue;
            }
        } 
    }
    delete[] users;
    users = nullptr;
    close(pipeFd);
    close(m_epollFd);
}

template <typename T>
void processPool<T>::runParent() {
    setUpSigPipe();
    
    addFd(m_epollFd, m_listenFd);

    epoll_event events[MAX_EVENT_NUMBER];
    int subProcessCounter = 0;
    int newConn = 0;
    int ret = -1;

    while(!m_stop) {
        number = epoll_wait(m_epollFd, events, MAX_EVENT_NUMBER, -1);
        if(number < 0 && errno != EINTR) {
            cout << "epoll failure" << endl;
            break;
        }
        for(int i = 0; i < number; ++i) {
            int sockFd = events[i].data.fd;
            if(sockFd == m_listenFd) {
                int i = subProcessCounter;
                do {
                    if(m_sub_process[i].m_pid != -1) {
                        break;
                    }
                    i = (i + 1) % m_process_number;
                }while(i != subProcessCounter);

                if(m_sub_process[i].m_pid == -1) {
                    m_stop = true;
                    break;
                }
                subProcessCounter = (i + 1)%m_process_number;
                send(m_sub_process[i].m_pipeFd[0], (char*)&newConn, sizeof(newConn), 0);
                cout << "send request to child " << i << endl;
            }
            else if (sockFd == sigPipeFd[0] && events[i].events & EPOLLIN) {
                int sig;
                char signals[1024];
                ret = recv(sigPipeFd[0], signals, sizeof(signals), 0);
                if(ret < 0) {
                    continue;
                }
                else {
                    for(int i = 0; i < ret; ++i) {
                        switch(signals[i]) {
                            case SIGCHLD: {
                                pid_t pid;
                                int stat;
                                while(pid = waitpid(-1, &stat, WNOHANG) > 0) {
                                    for(int i = 0; i < m_process_number; ++i) {
                                        if(m_sub_process[i].pid == pid) {
                                            cout << "child " << i << " join" << endl;
                                            close(m_sub_process[i].m_pipeFd[0]);
                                            m_sub_process[i].m_pid = -1;
                                        }
                                    }
                                }
                                m_stop = true;
                                for(int i = 0; i < m_process_number; ++i) {
                                    if(m_sub_process[i].m_pid != -1) {
                                        m_stop = false;
                                    }
                                }
                                break;
                            }
                            case SIGTERM:
                            case SIGINT: {
                                cout << "kill all the child now" << endl;
                                for(int i = 0; i < m_process_number; ++i) {
                                    int pid = m_sub_process[i].m_pid;
                                    if(pid != -1) {
                                        kill(pid, SIGTERM);
                                    
                                    }
                                }
                                break;
                            }
                            default: {
                                break;
                            }
                        }
                    }
                }
            }
            else {
                continue;
            }
        }
    }
    close(m_epollFd);
}


#endif // PROCESSPOOL_HPP
